package com.gitee.ixtf.rsocket.verticle

import com.gitee.ixtf.core.kotlinx.readJson
import com.gitee.ixtf.rsocket.BROKER_ROUTE
import com.gitee.ixtf.rsocket.DISCONNECTED_ERROR
import com.gitee.ixtf.rsocket.broker.*
import com.gitee.ixtf.rsocket.kotlinx.payload
import com.gitee.ixtf.rsocket.kotlinx.remoteAddress
import com.gitee.ixtf.rsocket.kotlinx.route
import com.gitee.ixtf.rsocket.service.GetBrokerServiceRequest
import com.gitee.ixtf.rsocket.service.ListBrokerClientRequest
import com.gitee.ixtf.rsocket.service.ListBrokerServiceRequest
import io.cloudevents.CloudEvent
import io.netty.util.ReferenceCountUtil
import io.rsocket.*
import io.vertx.core.json.JsonArray
import io.vertx.core.json.JsonObject
import io.vertx.kotlin.core.json.jsonArrayOf
import io.vertx.kotlin.coroutines.dispatcher
import jakarta.inject.Inject
import kotlinx.coroutines.reactor.mono
import org.reactivestreams.Publisher
import reactor.core.publisher.Flux
import reactor.core.publisher.Mono

abstract class BrokerServerVerticle : RSocketServerVerticle(), CloudEventHandler {
  @Inject protected lateinit var serviceManager: ServiceManager
  @Inject protected lateinit var clientManager: ClientManager
  private val brokerRSocket: RSocket by lazy {
    object : RSocket {
      override fun metadataPush(payload: Payload): Mono<Void> =
          pick(payload)?.metadataPush(payload) ?: Mono.error(DISCONNECTED_ERROR)

      override fun requestResponse(payload: Payload): Mono<Payload> =
          pick(payload)?.requestResponse(payload) ?: Mono.error(DISCONNECTED_ERROR)

      override fun requestStream(payload: Payload): Flux<Payload> =
          pick(payload)?.requestStream(payload) ?: Flux.error(DISCONNECTED_ERROR)

      override fun fireAndForget(payload: Payload): Mono<Void> =
          pick(payload)?.fireAndForget(payload) ?: Mono.error(DISCONNECTED_ERROR)

      override fun requestChannel(payloads: Publisher<Payload>): Flux<Payload> =
          Flux.from(payloads).switchOnFirst { t, u ->
            t.get()?.let { pick(it)?.requestChannel(u) } ?: Flux.empty()
          }
    }
  }

  override fun accept(setup: ConnectionSetupPayload, sendingSocket: RSocket): Mono<RSocket> =
      mono(vertx.dispatcher()) {
        log.debug("${sendingSocket.remoteAddress()}")
        val cloudEvent = SetupRequest.decode(setup)
        val principal = SetupRequest.principal(cloudEvent, authProvider)
        val service = SetupRequest.decodeService(cloudEvent)
        if (service == null) clientManager.add(principal, sendingSocket, cloudEvent)
        else serviceManager.add(principal, service, sendingSocket, cloudEvent)
        brokerRSocket
      }

  protected open fun pick(payload: Payload): RSocket? {
    try {
      val split = payload.route().split(":")
      val rSocket = if (BROKER_ROUTE == split[0]) this else serviceManager.pick(split)
      if (rSocket != null) return rSocket
    } catch (t: Throwable) {
      log.error(t)
    }
    ReferenceCountUtil.safeRelease(payload)
    return null
  }

  override suspend fun requestResponse(event: CloudEvent): Payload =
      when (event.type) {
        GetBrokerServiceRequest.CLOUD_EVENT_TYPE ->
            JsonObject.mapFrom(serviceManager.getResponse(event.readJson())).payload()
        ListBrokerServiceRequest.CLOUD_EVENT_TYPE ->
            jsonArrayOf(serviceManager.listResponse()).payload()
        ListBrokerClientRequest.CLOUD_EVENT_TYPE -> JsonArray(clientManager.list()).payload()
        else -> TODO()
      }
}
